package com.dyna.chat.listener;

import com.alibaba.fastjson.JSON;
import com.dyna.chat.dao.MessageDao;
import com.dyna.chat.domain.dto.msg.MsgSendMessageDTO;
import com.dyna.chat.domain.entity.Message;
import com.dyna.chat.domain.resp.msg.ChatMessageResp;
import com.dyna.chat.domain.resp.ws.WSBaseResp;
import com.dyna.chat.service.ChatService;
import com.dyna.chat.service.WSAdapter;
import com.dyna.chat.service.WebSocketService;
import com.dyna.constants.TopicConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Slf4j
@Component
public class MsgConsumerListener {

    @Resource
    private ChatService chatService;
    @Resource
    private MessageDao messageDao;
    @Resource
    private WebSocketService webSocketService;

    @KafkaListener(topics = TopicConstant.SEND_MSG_TOPIC, groupId = "group1")
    public void consumerMsg(String msg) {
        MsgSendMessageDTO dto = JSON.parseObject(msg, MsgSendMessageDTO.class);
        Message message = messageDao.getById(dto.getMsgId());
        ChatMessageResp msgResp = chatService.getMsgResp(message, null);
        WSBaseResp<ChatMessageResp> wsBaseResp = WSAdapter.buildMsgSend(msgResp);
        webSocketService.sendToAllOnline(wsBaseResp);
    }

}
